package com.ruoyi.matrix.queue

import cn.hutool.system.HostInfo
import com.ruoyi.common.utils.spring.SpringUtils
import com.ruoyi.matrix.dim.EnvDim
import org.slf4j.LoggerFactory
import org.springframework.data.redis.core.StringRedisTemplate
import java.time.Duration
import java.time.LocalDateTime
import java.time.LocalTime
import java.time.format.DateTimeFormatter

/**
 * 时间和空间受限的队列
 * 按照时间和空间分片的队列
 * 比如1分钟，100个元素一个队列，其队列名为：
 *
 * 2023040515_749_132
 * 2023040515_749_133
 *
 * 这种队列主要用于批次处理，兼顾延迟和吞吐
 * 以及适配下游的流量控制
 *
 * @property env            环境，用于区分开发环境和生产环境
 * @property queue          队列名主要的部分
 * @property timeSplit      时间分片大小，默认为5秒，不能超过1个小时，如果为1秒，则批次是无延时的，除非qps限流
 * @property spaceSplit     空间分片大小，默认为100个，最大10000个
 * @property qps            限流qps，每秒处理多少个批次，默认为0，表示不限流
 */
class TimeAndSpaceLimitedQueue(
    private val env: String,
    private val queue: String,
    private val timeSplit: Long = 5,
    private val spaceSplit: Int = 100,
    private val qps: Int = 0
) {
    private val fatmoreLock get() = SpringUtils.getBean<FatmoreLock>("fatmoreLock")
    private val redisTemplate get() = SpringUtils.getBean(StringRedisTemplate::class.java)
    private val hostname = HostInfo().name
    private val timeSplitFormat = DateTimeFormatter.ofPattern("yyyyMMddHH")
    private val maxSpaceSplit = 10000
    private val redoTaskQueue = SimpleQueue(env, "${queue}.redo")
    private val logger = LoggerFactory.getLogger(TimeAndSpaceLimitedQueue::class.java)
    val taskSize get() = redisTemplate.opsForZSet().zCard(getTaskQueue())!!

    /**
     * 桶以及它对应的任务延时
     * 就是说这个桶应该在多长的时间后开始处理它
     */
    data class BucketWithTaskDelay(val bucket: String, val delay: Long)

    /**
     * 将一批任务item分桶入桶
     * 无延时，桶内item不关注延时
     *
     * 在分桶和入桶的同时，也同时入任务队列，指定延时时间
     * 虽然每一个元素都会重复一次入任务队列的动作，但是这省略了边界代码，相当于touch一下
     *
     * @param value
     * @return
     */
    fun addTaskItems(values: List<String>) {
        if (spaceSplit > maxSpaceSplit) throw RuntimeException("桶的数量不能超过${maxSpaceSplit}")
        val taskQueue = getTaskQueue()

        // 单个简单处理
        if (values.size == 1) {
            val bucketWithDelay = getBucketWithTaskDelay()
            val now = System.currentTimeMillis()
            redisTemplate.opsForZSet().add(bucketWithDelay.bucket, values.first(), now.toDouble())
            val taskTime = System.currentTimeMillis() + bucketWithDelay.delay
            redisTemplate.opsForZSet().add(taskQueue, bucketWithDelay.bucket, taskTime.toDouble())
            return
        }

        // 多个必须分片处理
        val valuesGroupByBucket = values.map { Pair(it, getBucketWithTaskDelay()) }.groupBy { it.second.bucket }
        var taskSize = 0
        valuesGroupByBucket.forEach { (_, valuesWithBucketAndDelay) ->
            val valuesInBucket = valuesWithBucketAndDelay.map { it.first }
            val bucketWithDelay = valuesWithBucketAndDelay.last().second        // 取最后一个的延时
            val now = System.currentTimeMillis()
            addMultiplePipeline(bucketWithDelay.bucket, valuesInBucket, now, 0)
            val taskTime = System.currentTimeMillis() + bucketWithDelay.delay
            redisTemplate.opsForZSet().add(taskQueue, bucketWithDelay.bucket, taskTime.toDouble())!!
            taskSize++
        }
    }

    /**
     * 单个任务item分桶入桶
     * @see addTaskItems
     *
     * @param value
     * @return
     */
    fun addTaskItem(value: String) = addTaskItems(listOf(value))

    /**
     * 将任务item加入需要redo的任务队列
     * 等待到期重新分桶
     *
     * @param value
     * @return
     */
    fun redoTaskItem(value: String, delay: Long) = redoTaskQueue.add(value, delay)

    /**
     * 将一批任务item将入需要redo的任务队列
     * 等待到期重新分桶
     *
     * @param values
     * @param delay
     */
    fun redoTaskItems(values: List<String>, delay: Long) = redoTaskQueue.adds(values, delay)

    /**
     * 将任务item从桶里面移除
     *
     * @param bucket
     * @param value
     */
    fun removeTaskItem(bucket: String, value: String) {
        redisTemplate.opsForZSet().remove(bucket, value)
    }

    /**
     * 将一批任务item从桶里面移除
     *
     * @param bucket
     * @param values
     */
    fun remoteTaskItems(bucket: String, values: List<String>) = values.forEach { removeTaskItem(bucket, it) }

    /**
     * 获取某个桶（任务）的所有item
     *
     * @param bucket
     * @return
     */
    fun getTaskItems(bucket: String): Set<String> {
        val size = redisTemplate.opsForZSet().zCard(bucket) ?: 0L
        if (size > spaceSplit) throw RuntimeException("桶内的item个数大于spaceSplit")
        if (size > maxSpaceSplit) throw RuntimeException("桶的数量不能超过${maxSpaceSplit}")
        return redisTemplate.opsForZSet().range(bucket, 0, -1)?.toSet() ?: setOf()
    }

    /**
     * 获取一个任务
     * 该任务对应一个桶的所有任务item
     *
     * @param nextDelay 延迟多长时间继续处理？默认为5分钟
     * @return
     */
    fun pickTask(nextDelay: Duration = Duration.ofMinutes(5)): String? {
        // 如果有qps限流，则需要延迟出队
        // 这里是是一旦限流生效，则最快速返回，不需要再走后续逻辑
        if (qps > 0 && getQps() >= qps) return null

        val taskItems = redisTemplate.opsForZSet().rangeWithScores(getTaskQueue(), 0, 0)
        if (taskItems.isNullOrEmpty()) return null
        val taskItem = taskItems.toTypedArray()[0]
        if (taskItem.value == null || taskItem.score == null) return null
        val task = taskItem.value!!
        val score = taskItem.score!!.toLong()
        if (score > System.currentTimeMillis()) return null

        delayTask(task, nextDelay.toMillis())
        if (!lockTask(task, nextDelay.toMillis())) return null

        // 如果有qps限流，则需要延迟出队
        // 因为是并发抢，有一部分会这里incQps会大于qps，就被限流了
        if (qps > 0 && incQps() > qps) {
            delayTask(task, Duration.ofSeconds(1).toMillis())
            unlockTask(task)
            return null
        }

        return task
    }

    /**
     * 移除一个任务
     * 在移除任务的同时，会删除该任务对应的桶
     *
     * @param task
     */
    fun removeTask(task: String) {
        redisTemplate.opsForZSet().remove(getTaskQueue(), task)
        redisTemplate.delete(task)
    }

    /**
     * 延迟一个任务
     *
     * @return
     */
    fun delayTask(task: String, delay: Long): Boolean {
        val ts = System.currentTimeMillis() + delay
        return redisTemplate.opsForZSet().add(getTaskQueue(), task, ts.toDouble()) ?: false
    }

    fun resetProcessTimes(key: String) = fatmoreLock.resetForIncrease(wrapName("times.${queue}.${key}"))

    fun incProcessTimes(key: String): Long = fatmoreLock.increase(wrapName("times.${queue}.${key}"))

    fun getProcessTimes(key: String): Long = fatmoreLock.getBeforeIncrease(wrapName("times.${queue}.${key}"))

    fun lockTask(key: String, timeout: Long): Boolean =
        fatmoreLock.lock(wrapName("lock.${queue}.${key}"), hostname, timeout)

    fun lockTask(key: String, by: String, timeout: Long): Boolean =
        fatmoreLock.lock(wrapName("lock.${queue}.${key}"), by, timeout)

    fun unlockTask(key: String) = fatmoreLock.unlock(wrapName("lock.${queue}.${key}"))

    fun removeAndUnlockTask(value: String) =
        removeTask(value).also { unlockTask(value) }.also { resetProcessTimes(value) }

    fun removeAndUnlockTaskWithKey(value: String, key: String) = removeTask(value).also { unlockTask(key) }

    // 同时入队多个item
    private fun addMultiplePipeline(queue: String, values: List<String>, score: Long, delay: Long = 1L): Int {
        val scoreDouble = score.toDouble()
        return redisTemplate.executePipelined { conn ->
            values.forEachIndexed { index, it ->
                conn.zSetCommands()
                    .zAdd(queue.toByteArray(), scoreDouble + ((index + 1) * delay), it.toByteArray())
            }
            null
        }.size
    }

    // 获取当前的桶以及它对应的任务延时
    private fun getBucketWithTaskDelay(): BucketWithTaskDelay {
        val datetimeNow = LocalDateTime.now()
        val nowSecondOfDay = LocalTime.now().toSecondOfDay()                // 时间片是每天滚动一次
        val timePart = (nowSecondOfDay + 1) / timeSplit
        val spacePart = incCount() / spaceSplit                             // 数量是全局累计
        val datePart = datetimeNow.format(timeSplitFormat)
        val bucket = "${env}.${queue}.bucket.${datePart}_${timePart}_${spacePart}"
        val delay = timeSplit - nowSecondOfDay % timeSplit                  // 这个延时类似于timeSplit倒计时
        return BucketWithTaskDelay(bucket, delay * 1000)
    }

    // 获取桶号队列（任务队列）
    private fun getTaskQueue(): String = wrapName("${queue}.task")
    private fun incCount() = redisTemplate.opsForValue().increment(wrapName(queue))!!

    // 获取当前秒的qps计数
    private fun getQps(): Long {
        val tsNow = System.currentTimeMillis() / 1000
        val key = wrapName("${queue}.qps.${tsNow}")
        return redisTemplate.opsForValue().get(key)?.toLong() ?: 0L
    }

    // 累计每秒的qps计数
    private fun incQps(): Long {
        val tsNow = System.currentTimeMillis() / 1000
        val key = wrapName("${queue}.qps.${tsNow}")
        val qpsCount = redisTemplate.opsForValue().increment(key)
        redisTemplate.expire(key, Duration.ofSeconds(3))
        return qpsCount!!
    }

    // 开发环境用私人hostname隔离
    private fun wrapName(queue: String): String {
        return if (env == EnvDim.prod) "${env}.${queue}"
        else "${env}.${hostname}.${queue}"
    }

    private val interval = 20L
    fun startRedoTask() {
        logger.info("[${queue}] 启动redoTask")
        while (true) {
            try {
                val item = redoTaskQueue.pick()
                if (item == null) {
                    Thread.sleep(interval)
                    continue
                }
                if (item.score > System.currentTimeMillis()) {
                    Thread.sleep(interval)
                    continue
                }
                val nextDelay = Duration.ofMinutes(5).toMinutes()
                if (!redoTaskQueue.lock(item.member, nextDelay)) {
                    Thread.sleep(interval)
                    continue
                }
                redoTaskQueue.delay(item.member, nextDelay)
                addTaskItem(item.member)
                redoTaskQueue.removeAndUnlock(item.member)
            } catch (e: Exception) {
                logger.error("redo异常", e)
            }
            Thread.sleep(interval)
        }
    }
}